package com.yanggu.flink.datastream_api.transform

import org.apache.flink.streaming.api.scala._

/**
 * keyBy之后使用简单聚合算子
 * 例如sum、max、min等
 */
object TransTupleAggregationDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.fromElements(("a", 1), ("a", 3), ("b", 3), ("b", 4))

    //DataStream  =>  KeyedStream  =>  DataStream
    //          keyBY后           聚合之后
    //返回的流中包含每一次聚合的结果, 而不是只返回最后一次聚合的最终结果
    //因为Flink是流式计算, 来一条数据, 也就会计算一次, 进而输出一次
    stream.keyBy(_._1).sum(1).print("sum position>")
    stream.keyBy(_._1).sum("_2").print("sum field>")
    stream.keyBy(_._1).max(1).print("max position>")
    stream.keyBy(_._1).max("_2").print("max field>")
    stream.keyBy(_._1).min(1).print("min position>")
    stream.keyBy(_._1).min("_2").print("min field>")
    stream.keyBy(_._1).maxBy(1).print("maxBy position>")
    stream.keyBy(_._1).maxBy("_2").print("maxBy field>")
    stream.keyBy(_._1).minBy(1).print("minBy position>")
    stream.keyBy(_._1).minBy("_2").print("minBy field>")

    env.execute()
  }

}
